package com.ksc.wordcount.datasourceapi;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

/**
 * 自定义文件格式实现类，用于处理不可切分的文件。
 */
public class SplitFileFormat implements FileFormat {

    /**
     * 判断文件是否可切割
     *
     * @param filePath 文件路径
     * @return 是否可切割，此处为不可切割，返回false
     */
    @Override
    public boolean isSplitable(String filePath) {
        return true;
    }

    /**
     * 获取文件的切割块
     *
     * @param filePath 文件路径
     * @param size     文件大小
     * @return 文件的切割块数组
     */
    @Override
    public PartionFile[] getSplits(String filePath, long size) {
        File parentFile = new File(filePath);
        if (parentFile.isFile()) {
            // 如果是单个文件，直接返回一个PartionFile，其中包含一个FileSplit
            return new PartionFile[]{new PartionFile(0, new FileSplit[]{new FileSplit(filePath, 0, parentFile.length())})};
        }
        List<PartionFile> partitionFileList = new ArrayList<>();
        /////todo 实现 driver端切分split的逻辑
        // 如果是目录，遍历目录下的文件，并将每个文件生成一个PartionFile，其中包含一个FileSplit
        File[] files = parentFile.listFiles();
        int partitionId = 0;
        for (File file : files) {
            try (BufferedReader reader = new BufferedReader(new FileReader(file))) {
                String line;
                long position = 0;
                long remainingSize = file.length();
                StringBuilder buffer = new StringBuilder();
                while ((line = reader.readLine()) != null) {
                    long lineSize = line.getBytes().length;
                    int bufferSize = buffer.length();
                    long totalSize=bufferSize+lineSize;
                    if (totalSize <= size) {
                        // 拼接到当前分片
                        // 每一行最后面的空格也占据两个字符不等，因此也需要添加上
                        buffer.append(line).append(System.lineSeparator());
                    } else if (remainingSize <= size) {
                        // 将剩余的部分作为一个切分
                        FileSplit fileSplit = new FileSplit(file.getAbsolutePath(), position, (position + remainingSize - 1));
                        partitionFileList.add(new PartionFile(partitionId, new FileSplit[]{fileSplit}));
                        partitionId++;
                        break;
                    } else {
                        // 生成当前分片并加入列表
                        buffer.append(System.lineSeparator());
                        long blockSize = lineSize + buffer.toString().getBytes().length;
                        long end = position + blockSize ;
                        FileSplit fileSplit = new FileSplit(file.getAbsolutePath(), position, end);
                        partitionFileList.add(new PartionFile(partitionId, new FileSplit[]{fileSplit}));
                        partitionId++;
                        position += blockSize;
                        remainingSize -= blockSize;
                        // 清空缓冲区并将当前行加入下一个分片
                        buffer.setLength(0);
                    }
                }

                if (buffer.length() > 0) {
                    long end = position + buffer.toString().getBytes().length ;
                    // 生成最后一个分片并加入列表
                    FileSplit fileSplit = new FileSplit(file.getAbsolutePath(), position, end);
                    partitionFileList.add(new PartionFile(partitionId, new FileSplit[]{fileSplit}));
                    partitionId++;
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        // 将生成的PartionFile转换为PartionFile数组返回
        return partitionFileList.toArray(new PartionFile[partitionFileList.size()]);
    }

    /**
     * 创建读取器
     *
     * @return PartionReader实例，此处返回TextPartionReader实例
     */
    @Override
    public PartionReader createReader() {
        return new TextPartionReader();
    }

    /**
     * 创建写入器
     *
     * @param destPath  写入目标路径
     * @param partionId 切割块ID
     * @return PartionWriter实例，此处返回TextPartionWriter实例
     */
    @Override
    public PartionWriter createWriter(String destPath, int partionId,String applicationId) {
        return new TextPartionWriter(destPath, partionId,applicationId);
    }



//    public static PartionFile[] getSplitstest(String filePath, long size) {
//        File parentFile = new File(filePath);
//        if (parentFile.isFile()) {
//            // 如果是单个文件，直接返回一个PartionFile，其中包含一个FileSplit
//            return new PartionFile[]{new PartionFile(0, new FileSplit[]{new FileSplit(filePath, 0, parentFile.length())})};
//        }
//        List<PartionFile> partitionFileList = new ArrayList<>();
//        /////todo 实现 driver端切分split的逻辑
//        // 如果是目录，遍历目录下的文件，并将每个文件生成一个PartionFile，其中包含一个FileSplit
//        File[] files = parentFile.listFiles();
//        int partitionId = 0;
//        for (File file : files) {
//            try (BufferedReader reader = new BufferedReader(new FileReader(file))) {
//                String line;
//                long position = 0;
//                long remainingSize = file.length();
//                StringBuilder buffer = new StringBuilder();
//                while ((line = reader.readLine()) != null) {
//                    long lineSize = line.getBytes().length;
//                    int bufferSize = buffer.length();
//                    long totalSize=bufferSize+lineSize;
//                    if (totalSize <= size) {
//                        // 拼接到当前分片
//                        // 每一行最后面的空格也占据两个字符不等，因此也需要添加上
//                        buffer.append(line).append(System.lineSeparator());
//                    } else if (remainingSize <= size) {
//                        // 将剩余的部分作为一个切分
//                        FileSplit fileSplit = new FileSplit(file.getAbsolutePath(), position, (position + remainingSize - 1));
//                        partitionFileList.add(new PartionFile(partitionId, new FileSplit[]{fileSplit}));
//                        partitionId++;
//                        break;
//                    } else {
//                        // 生成当前分片并加入列表
//                        buffer.append(System.lineSeparator());
//                        long blockSize = lineSize + buffer.toString().getBytes().length;
//                        long end = position + blockSize ;
//                        FileSplit fileSplit = new FileSplit(file.getAbsolutePath(), position, end);
//                        partitionFileList.add(new PartionFile(partitionId, new FileSplit[]{fileSplit}));
//                        partitionId++;
//                        position += blockSize;
//                        remainingSize -= blockSize;
//                        // 清空缓冲区并将当前行加入下一个分片
//                        buffer.setLength(0);
//                    }
//                }
//
//                if (buffer.length() > 0) {
//                    long end = position + buffer.toString().getBytes().length ;
//                    // 生成最后一个分片并加入列表
//                    FileSplit fileSplit = new FileSplit(file.getAbsolutePath(), position, end);
//                    partitionFileList.add(new PartionFile(partitionId, new FileSplit[]{fileSplit}));
//                    partitionId++;
//                }
//            } catch (IOException e) {
//                e.printStackTrace();
//            }
//        }
//
//        System.out.println("partitionFileList is ============");
//        for (PartionFile file :partitionFileList) {
//            int partionId = file.getPartionId();
//            FileSplit[] fileSplits = file.getFileSplits();
//            System.out.println(partionId);
//            for (FileSplit fileSplit:fileSplits) {
//                try (BufferedReader reader = new BufferedReader(new FileReader(fileSplit.getFileName()))) {
//                    reader.skip(fileSplit.getStart());
//                    char[] buffer = new char[(int) (fileSplit.getLength()- fileSplit.getStart())];
//                    int bytesRead = reader.read(buffer);
//                    String content = new String(buffer, 0, bytesRead);
//                    System.out.println(new File(fileSplit.getFileName()).getName() + "的文件分片内容如下: ");
//                    System.out.println(content);
//                } catch (IOException e) {
//                    e.printStackTrace();
//                }
//            }
//
//        }
//
//        // 将生成的PartionFile转换为PartionFile数组返回
//        return partitionFileList.toArray(new PartionFile[partitionFileList.size()]);
//    }

//    private static void printFileContent(FileSplit fileSplit) {
//        try (BufferedReader reader = new BufferedReader(new FileReader(fileSplit.getFileName()))) {
//            reader.skip(fileSplit.getStart());
//            char[] buffer = new char[(int) (fileSplit.getLength()- fileSplit.getStart())];
//            int bytesRead = reader.read(buffer);
//            String content = new String(buffer, 0, bytesRead);
//            System.out.println(new File(fileSplit.getFileName()).getName() + "的文件分片内容如下: ");
//            System.out.println(content);
//        } catch (IOException e) {
//            e.printStackTrace();
//        }
//    }

//    public static void main(String[] args) {
//        String filePath = "D:\\kingsoft\\source\\TopN\\wordcountdemo2\\tmp\\input";
//        long splitSize = 100;
//
//        PartionFile[] splits = getSplitstest(filePath, splitSize);
//
//
//        // 打印切分后的各个分片文件内容
////        for (PartionFile partionFile : splits) {
////            for (FileSplit fileSplit : partionFile.getFileSplits()) {
////                printFileContent(fileSplit);
////            }
////        }
//    }
//    public static void test(){
//        String filePath = "D:\\kingsoft\\source\\TopN\\wordcountdemo2\\tmp\\input\\inputFile1.log";
//        long splitSize = 100;
//
//        try (BufferedReader reader = new BufferedReader(new FileReader(filePath))) {
//            reader.skip(0);
//            char[] buffer = new char[151];
//            int bytesRead = reader.read(buffer);
//            String content = new String(buffer, 0, bytesRead);
//            System.out.println("File Content for " + ": " + content);
//        } catch (IOException e) {
//            e.printStackTrace();
//        }
//    }

}